package realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

public class DwdExamQuestion {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_exam_question_group"));
        Table examQuestion = tableEnv.sqlQuery("select\n" +
                "    `data`['id'] id,\n" +
                "    `data`['exam_id'] exam_id,\n" +
                "    `data`['paper_id'] paper_id,\n" +
                "    `data`['question_id'] question_id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['is_correct'] is_correct,\n" +
                "    `data`['score'] score,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    ts\n" +
                "from\n" +
                "    topic_db\n" +
                "where \n" +
                "    `table`='test_exam_question' and `type`='insert'");
        tableEnv.createTemporaryView("exam_question",examQuestion);

        tableEnv.executeSql("CREATE TABLE dwd_test_exam_question (\n" +
                "    id string,\n" +
                "    exam_id string,\n" +
                "    paper_id string,\n" +
                "    question_id string,\n" +
                "    user_id string,\n" +
                "    is_correct string,\n" +
                "    score string,\n" +
                "    create_time string,\n" +
                "    ts string,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + KafkaUtil.getUpsertKafkaDDL("dwd_test_exam_question"));
        tableEnv.executeSql("insert into dwd_test_exam_question select * from exam_question");
    }
}
